package each_webcode

import common.HbaseTools
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.util.parsing.json.JSON;
object IBRD_Not_Change {
def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName("IBRDdatachange")//.setMaster("local[*]") //集群运行去掉setMaster
        val sc = new SparkContext(sparkConf)
        sc.setLogLevel("Error")
        //val baseInfotable: String = new String("BaseMapInfos")
        //val detailTable: String = new String("DetailMapInfos")

        //val inpath = args(0)
        //val outpath = args(1)
        //    val inpath = "/Users/kelisiduofu/IdeaProjects/zhiwang/InternationalData/makeChange/src/main/resources/sub_IBRD_Value.json"
        //    val outpath = "/Users/kelisiduofu/IdeaProjects/zhiwang/InternationalData/makeChange/src/main/resources/result"

        val inpath="hdfs://node01:8020/tmp/InternationalData_source/InternationalData.IBRD_Value.json"
        val outpath ="hdfs://node01:8020/tmp/InternationalData_Result/IBRD"
        val jsonStr = sc.textFile(inpath)

        val lines: RDD[Option[Any]] = jsonStr.map(one_line => JSON.parseFull(one_line))

        val pre_rdd = lines.map {
                r =>
                        r match {
                                case Some(maps: Map[String, Any]) => {
                                        val my_map: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()
                                        maps.foreach(t => {
                                                my_map += (t._1 -> t._2.toString.replace("\n", ""))
                                                })
                                        my_map("_id") = my_map("_id").substring(12, 36)
                                        //(scala.util.Random.nextInt(200).toString + "_" + "IBRD_" + my_map("SourceCode"), my_map)
                                        (scala.util.Random.nextInt(200).toString+"_"+"IBRD_" + my_map("SourceCode"), my_map)
                                        }
                                }
                }

        val result_rdd: RDD[String] = pre_rdd.groupByKey(400).map {
                case (k: String, v) => {
                        val arr = k.split("_")
                        //val k1=arr(1)+"_"+arr(2)
                                val k1 = arr(1) + "_" + arr(2)
                        val tools = new HbaseTools()
                        val connection = tools.getConnection()
                        //dealTypes:map(kbase表的列名,列对应type类型)
                                val dealTypes: Map[String, String] = tools.getFamily_column_value(connection, "BaseMapInfos", k1, "dealTypes").toMap
                        //map(kbase列名,value表列名)
                                val map01: Map[String, String] = tools.getFamily_column_value(connection, "BaseMapInfos", k1, "map01").toMap
                        //map(conceptId,map(detailId,DetailName))
                                //map("WebInfo",map(SourceCode,值))
                                //map("WebInfo",map(SourceName,值))
                                //map("WebInfo",map(WebCode,值))
                                //map("WebInfo",map(WebName,值))
                                val map04 = new mutable.HashMap[String, mutable.Map[String, String]]()
                        val familys = tools.getFamilys_list(connection, "DetailMapInfos", k1)
                        for (family: String <- familys) {
                                //列名,列值
                                        val tmp: mutable.HashMap[String, String] = tools.getFamily_column_value(connection, "DetailMapInfos", k1, family)
                                map04.put(family, tmp)
                                }

                        //map(detailname,detailId)
                                val PU_timeMap: Map[String, String] = tools.getFamily_column_value(connection, "BaseMapInfos", "public", "public_Time").toMap
                        //make_Dic,map(detailname,detailId)
                                val PU_country: Map[String, String] = tools.getFamily_column_value(connection, "BaseMapInfos", "public", "public_country").toMap

                        connection.close()
                        val result: Iterable[String] = v.map {
                                keys_values => {
                                        val result_map = new mutable.HashMap[String, String]()
                                        val map03 = keys_values
                                        dealTypes.keys.foreach {
                                                key =>
                                                        dealTypes(key) match {
                                                                case "type01" => { //map01+map03
                                                                        if (map01.contains(key) && map03.contains(map01(key))) {
                                                                                //SourceCode=29,43,2   Year Time
                                                                                //29:Time格式为YR2015
                                                                                if(map03("SourceCode")=="29" && key=="Year" && map03(map01(key)).length==6)
                                                                                        result_map.put(key, map03(map01(key)).split("R")(1))
                                                                                        //29:Time格式为YR2015-28
                                                                                        else if(map03("SourceCode")=="29" && key=="Year" && map03(map01(key)).length==9)
                                                                                        result_map.put(key, map03(map01(key)).split("-")(0).split("R")(1))
                                                                                        //54:Time格式为YR1990Q1  Year取1008
                                                                                        else if(map03("SourceCode")=="43" && key=="Year" && map03(map01(key)).length==8){
                                                                                        result_map.put(key, map03(map01(key)).split("R")(1).split("Q")(0))
                                                                                        //此时加入Period字段,取Q1
                                                                                        result_map.put("Period", "Q"+map03(map01(key)).split("R")(1).split("Q")(1))
                                                                                        }
                                                                                        //或许有YR2015这样的格式
                                                                                        else if(map03("SourceCode")=="43" && key=="Year" && map03(map01(key)).length==6){
                                                                                        result_map.put(key, map03(map01(key)).split("R")(1))
                                                                                        }
                                                                                        //2:Time格式为YR1990
                                                                                        else if(map03("SourceCode")=="2" && key=="Year" && map03(map01(key)).length==6){
                                                                                        result_map.put(key, map03(map01(key)).split("R")(1))
                                                                                        }
                                                                                        //2:Time格式为YR1990Q1 Year取1008
                                                                                        else if(map03("SourceCode")=="2" && key=="Year" && map03(map01(key)).length==8){
                                                                                        result_map.put(key, map03(map01(key)).split("R")(1).split("Q")(0))
                                                                                        //此时加入Period字段,取Q1
                                                                                        result_map.put("Period", "Q"+map03(map01(key)).split("R")(1).split("Q")(1))
                                                                                        }
                                                                                        else
                                                                                        result_map.put(key, map03(map01(key)))
                                                                                }
                                                                        }
                                                                        case "type02" => {
                                                                                result_map.put(key, map04("WebInfo")(map01(key)))
                                                                                }
                                                                        case "type05" => {
                                                                                if (map01.contains(key) && map03.contains(map01(key)) && map04.contains(map01(key)) && map04(map01(key)).contains(map03(map01(key)))) {
                                                                                        //SourceCode=54,中的Year含有Q季度,切掉后面的Q季度,只保留年
                                                                                        if (map03("SourceCode")=="54" && key=="Year")
                                                                                                result_map.put(key, map04(map01(key))(map03(map01(key))).split("Q")(0))
                                                                                                //SourceCode=54,格式为2013 Targe或2013,切掉后面的Q季度,只保留年
                                                                                                else if (map03("SourceCode")=="34" && key=="Year")
                                                                                                result_map.put(key, map04(map01(key))(map03(map01(key))).split(" ")(0))
                                                                                                else
                                                                                                result_map.put(key, map04(map01(key))(map03(map01(key))))
                                                                                        }
                                                                                }
                                                                        //SourceCode=15 20 22 23 67 68 76 77
                                                                        //Year Time 中的年度
                                                                        //Priod Time 中的季度
                                                                        case "type01_bird" => {
                                                                                if (map03(map01("Period")).length == 8) {
                                                                                        val time = map03(map01("Period"))
                                                                                        result_map.put("Year", time.substring(2, 6))
                                                                                        result_map.put("Period", time.substring(6, 8))
                                                                                        }

                                                                                if (map03(map01("Period")).length == 6) {
                                                                                        val time = map03(map01("Period"))
                                                                                        result_map.put("Year", time.substring(0, 4))
                                                                                        result_map.put("Period", time.substring(4, 6))
                                                                                        }

                                                                                if (map03(map01("Period")).length == 7) {
                                                                                        val time = map03(map01("Period"))
                                                                                        result_map.put("Year", time.substring(0, 4))
                                                                                        result_map.put("Period", time.substring(4, 7))
                                                                                        }
                                                                                if (map03(map01("Period")).length == 4) {
                                                                                        val time = map03(map01("Period"))
                                                                                        result_map.put("Year", time)
                                                                                        result_map.put("Period", "")
                                                                                        }
                                                                                }
                                                                        case "type02_bird" => {
                                                                                if (map01.contains(key) && map03.contains(map01(key)) && map04.contains(map01(key)) && map04(map01(key)).contains(map03(map01(key)))) {
                                                                                        result_map.put(key, "Version:" + map04(map01(key))(map03(map01(key))))
                                                                                        }
                                                                                }
                                                                        //处理map03 key:'Receiving Countries ',多了一个空格,ReporterCode
                                                                        case "type03_bird" => { //map01+map03
                                                                                if (map01.contains(key) && map03.contains(map01(key) + " ")) {
                                                                                        result_map.put(key, map03(map01(key) + " "))
                                                                                        }
                                                                                }
                                                                        //处理map03 key:'Receiving Countries ',Reporter
                                                                        case "type04_bird" => { //map01+map03
                                                                                if (map01.contains(key) && map03.contains(map01(key) + " ") && map04.contains(map01(key)) && map04(map01(key)).contains(map03(map01(key) + " "))) {
                                                                                        result_map.put(key, map04(map01(key))(map03(map01(key) + " ")))
                                                                                        }
                                                                                }
                                                                //处理SourceCode=54,原为Year Time Time,改为Year Time Time中Q前面的字段
                                                                }
                                                }

                                        //生成TimeSort字段，
                                        if (result_map.contains("Year") && result_map.contains("Period") && result_map("Period").trim != "") {
                                                if (PU_timeMap.contains(result_map("Period").trim)) {
                                                        result_map.put("TimeSort", result_map("Year") + PU_timeMap(result_map("Period").trim))
                                                        } else {
                                                        result_map.put("TimeSort", result_map.getOrElse("Year", ""))
                                                        }
                                                } else {
                                                result_map.put("TimeSort", result_map.getOrElse("Year", ""))
                                                }

                                        //生成FuzzyIndicator字段
                                        if (result_map.contains("Indicator") && result_map("Indicator") != "") {
                                                result_map.put("FuzzyIndicator", result_map("Indicator"))
                                                //生成IndicatorLength字段
                                                result_map.put("IndicatorLength", result_map("Indicator").length.toString)
                                                }
                                        //生成FuzzyReporter字段
                                        if (result_map.contains("Reporter") && result_map("Reporter") != "") {
                                                if (PU_country.contains(result_map("Reporter"))) {
                                                        result_map.put("FuzzyReporter", result_map("Reporter") + "，" + PU_country(result_map("Reporter")))
                                                        } else {
                                                        result_map.put("FuzzyReporter", result_map("Reporter"))
                                                        }
                                                }

                                        val MetaData: Array[String] = Array("WebCode", "SourceCode", "ReporterCode", "Reporter", "ParternerCode", "Parterner", "Year", "Period",
                                                "IndicatorCategory", "Indicator", "Dim1Type", "Dim1", "Dim2Type", "Dim2", "Dim3Type", "Dim3",
                                                "Dim4Type", "Dim4", "Dim5Type", "Dim5", "Dim6Type", "Dim6", "Dim7Type", "Dim7", "Dim8Type", "Dim8")
                                        val DIMTYPEMETADATA: Array[String] = Array("Dim1Type", "Dim2Type", "Dim3Type", "Dim4Type", "Dim5Type", "Dim6Type", "Dim7Type", "Dim8Type")
                                        val DIMMETADATA: Array[String] = Array("Dim1", "Dim2", "Dim3", "Dim4", "Dim5", "Dim6", "Dim7", "Dim8")

                                        //生成DIMTYPEMETADATA:各Dim1Type值的组合,存在的为"值，"不存在的去掉
                                                val dimtypemetadata = DIMTYPEMETADATA.flatMap(result_map.getOrElse(_, "<->") + "，").mkString("").replace("<->，", "")

                                        //生成DIMMETADATA：各Dim组合，存在的为"值，"不存在的去掉
                                                val dimemetadata = DIMMETADATA.flatMap(onecel => {
                                                result_map.getOrElse(onecel, "<->") + "，"
                                                }).mkString("").replace("<->，", "")
                                        //生成MetaData：部分字段值的组合,存在的为"值，"不存在的去掉
                                                val metadata = MetaData.flatMap(result_map.getOrElse(_, "<->") + "，").mkString("").replace("<->，", "")
                                        result_map.put("DIMTYPEMETADATA", dimtypemetadata)
                                        result_map.put("DIMMETADATA", dimemetadata)
                                        result_map.put("MetaData", metadata)
                                        //val kkl:RDD[String] = sc.parallelize(().toString)
                                        //将map(key -> value)间的分隔符改成\t
                                        result_map.mkString("(<>)").replace(" -> ", "\t")
                                        }
                                }

                        val res = result.mkString("\n")
                        res
                        }
                }

        //val result_rdd: RDD[String]

        // 按照换行符切割
                val lines2: RDD[String] = result_rdd.flatMap(lines => {
                lines.split("\n")
                })

        val pangpang: RDD[String] = lines2.map(line => {
                // 按照\(<>\)切割
                        val strs: Array[String] = line.split("\\(<>\\)")

                val m = new mutable.HashMap[String, String]()

                strs.map(x => {
                        // 按照\t切割
                                val s = x.split("\t")
                        // 长度不为2,放入空串
                        if (s.length < 2) {
                                m.put(s(0), "")
                                }
                                else
                                // 长度为2,正常放入集合中
                                m.put(s(0), s(1))
                        })
                val res1: Map[String, String] = m.toMap
                // 过滤掉为空串的
                        val res2 = res1.filter(x => {
                        var flag: Boolean = true
                        // 如果集合中不存在ObservationValue所对应的value值,或者value值为"",则返回false,过滤掉
                        if (m.getOrElse("ObservationValue", "").equals("")||m.getOrElse("Indicator", "").equals(""))
                                flag = false
                        flag
                        })
                // 将集合进行拼接
                res2.mkString("(<>)").replace(" -> ", "\t")
                })


        pangpang.filter(!_.equals("")).saveAsTextFile(outpath)
        sc.stop()
        }
}
